package com.techsun.datanow.sync

import com.microsoft.azure.cosmosdb.spark.config.Config
import com.microsoft.azure.cosmosdb.spark.schema._
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._


class Sync_Order extends TDataSync {
  /**
   * 同步工作方法
   *
   * @param spark             spark 会话对象
   * @param cosmosDbEndpoint  CosmosDB Endpoint
   * @param cosmosDbMasterkey CosmosDB Endpoint 对应的密钥
   * @param kuduMaster        kudu Master 节点
   * @return 同步成功或失败
   */
  override def sync(spark: SparkSession, cosmosDbEndpoint: String, cosmosDbMasterkey: String, kuduMaster: String, beginTime: Long, endTime: Long): Boolean = {
    try {
      Logger.getLogger("org").setLevel(Level.ERROR)
      Logger.getLogger("com").setLevel(Level.ERROR)
      val tableName = "sh_tsrorder"
      val config = Config(Map("Endpoint" -> cosmosDbEndpoint,
        "Masterkey" -> cosmosDbMasterkey,
        "Database" -> "VF_Vans_CN",
        "preferredRegions" -> "China East 2;",
        "Collection" -> "TsrOrder",
        "query_custom" ->
          s""" SELECT c.id id, c.code code, c.storeId storeid, c.storeCode storecode, c.storeName storename,
             | c.orderDate AS orderdate, c.source source, c.transactionCode transactioncode, c.direction direction, c.orderType ordertype,c.payTime AS paytime, c.customerId customerid,
             | c.memberId memberid,  c.orderStatus orderstatus, c.totalPrice AS totalprice, c.totalAmount as totalamount,
             | c.couponAmount AS couponamount, c.discountAmount AS discountamount, c.totalDisAmount totaldisamount, c.costAmount as costamount, c.grossAmount AS grossamount,
             | c.qty qty, c.salesId AS salesid, c.originalId AS originalid, c.remark remark, c.deliveryTime AS deliverytime,  c.buyerName AS buyername,
             | c.freight AS freight, c.receiverName AS receivername, c.receiverMobile AS receivermobile, c.receiverTel AS receivertel, c.province AS province,
             |  c.city AS city, c.area AS area, c.address AS address, c.zipcode AS zipcode, c.calculateStatus calculatestatus,c.createTime AS createtime, c.updateTime AS updatetime, c._ts as _ts,c.transactionDate transactiondate
             |  FROM c
             | where c._ts >= ${beginTime} """.stripMargin))

      val schema = StructType(
        List(
          StructField("id", StringType, true),
          StructField("memberid", StringType, true),
          StructField("customerid", StringType, true),
          StructField("code", StringType, true),
          StructField("storeid", StringType, true),
          StructField("storecode", StringType, true),
          StructField("storename", StringType, true),
          StructField("orderdate", TimestampType, true),
          StructField("paytime", TimestampType, true),
          StructField("source", IntegerType, true),
          StructField("transactioncode", StringType, true),
          StructField("direction", IntegerType, true),
          StructField("ordertype", IntegerType, true),
          StructField("orderstatus", IntegerType, true),
          StructField("totalprice", DoubleType, true),
          StructField("totalamount", DoubleType, true),
          StructField("couponamount", DoubleType, true),
          StructField("discountamount", DoubleType, true),
          StructField("totaldisamount", DoubleType, true),
          StructField("costamount", DoubleType, true),
          StructField("grossamount", DoubleType, true),
          StructField("qty", IntegerType, true),
          StructField("salesid", StringType, true),
          StructField("originalid", StringType, true),
          StructField("remark", StringType, true),
          StructField("deliverytime", TimestampType, true),
          StructField("buyername", StringType, true),
          StructField("freight", DoubleType, true),
          StructField("receivername", StringType, true),
          StructField("receivermobile", StringType, true),
          StructField("receivertel", StringType, true),
          StructField("province", StringType, true),
          StructField("city", StringType, true),
          StructField("area", StringType, true),
          StructField("address", StringType, true),
          StructField("zipcode", StringType, true),
          StructField("calculatestatus", IntegerType, true),
          StructField("createtime", TimestampType, true),
          StructField("updatetime", TimestampType, true),
          StructField("_ts", IntegerType, true),
          StructField("transactiondate", TimestampType, true)
        ))
      val df = spark.sqlContext.read.schema(schema).cosmosDB(config)
      val count = df.count();
      printf("Sync_TsrOrder query count: %d\n", count)
      val kuduContext = new KuduContext(kuduMaster, spark.sparkContext)
      kuduContext.upsertRows(df, tableName)
      return true
    } catch {
      case ex: Exception => {
        println("Sync_TsrOrder exception: " + ex.getMessage)
        return false
      }
    }
  }
}
